package com.example.java_message_queue.mqserver.datacenter;

import com.example.java_message_queue.common.BinaryTool;
import com.example.java_message_queue.common.MqException;
import com.example.java_message_queue.mqserver.core.MSGQueue;
import com.example.java_message_queue.mqserver.core.Message;

import java.io.*;
import java.util.LinkedList;
import java.util.List;
import java.util.Scanner;

/*
 * 文件格式约定:
 *
 * queue_data.txt
 * 保存所有的消息数据
 * 每个 Message 对象, 通过 Java 标准库的序列化方法进行序列化.
 * 在写入文件的时候, 先给文件中写入 Message 对象的长度(4个字节), 然后再写入对象的数据本体. 其中 offsetBeg 和 offsetEnd 不参与序列化
 * 这样设定主要是为了后续 recovery 这里的功能支持.
 * 而对于实现消息的发送和消费, 则不依赖文件中的长度. 内存中直接通过 offsetBeg 和 offsetEnd 定位到文件中指定对象, 读取, 修改, 重新写入即可.
 *
 * queue_stat.txt
 * 简单的行文本文件. 里面只有一行. 这一行里有两列, 使用 \t 分割.
 * 第一列是总的消息数; 第二列是有效消息数.
 */
public class MessageFileManager {
    // 表示消息的统计信息
    static public class Stat {
        public int totalCount;
        public int validCount;
    }

    public void init() {
        // 当前这里不需要做任何工作.
    }

    // 队列目录
    private String getQueueDir(String queueName) {
        return "./data/" + queueName;
    }

    // 队列数据文件
    // 这个文件来存储队列的真实数据
    private String getQueueDataPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data.txt";
    }

    // 队列统计文件
    // 这个文件用来存储队列中的统计信息.
    // 包含一行, 两个列使用 \t 分割, 分别是总数据, 和无效数据.
    private String getQueueStatPath(String queueName) {
        return getQueueDir(queueName) + "/queue_stat.txt";
    }

    public void createQueueFiles(String queueName) throws IOException {
        // 1. 创建目录指定队列的目录
        File baseDir = new File(getQueueDir(queueName));
        if (!baseDir.exists()) {
            boolean ok = baseDir.mkdirs();
            if (!ok) {
                throw new IOException("创建目录失败! baseDir=" + baseDir.getAbsolutePath());
            }
        }
        // 2. 创建队列数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            boolean ok = queueDataFile.createNewFile();
            if (!ok) {
                throw new IOException("创建文件失败! queueDataFile=" + queueDataFile.getAbsolutePath());
            }
        }
        // 3. 创建队列统计文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            boolean ok = queueStatFile.createNewFile();
            if (!ok) {
                throw new IOException("创建文件失败! queueStatFile=" + queueStatFile.getAbsolutePath());
            }
        }
        // 4. 给队列统计文件写入初始数据
        Stat stat = new Stat();
        stat.totalCount = 0;
        stat.validCount = 0;
        writeStat(queueName, stat);
    }

    public void destroyQueueFiles(String queueName) throws IOException {
        // 1. 先删除目录中的文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        boolean ok1 = queueDataFile.delete();
        File queueStatFile = new File(getQueueStatPath(queueName));
        boolean ok2 = queueStatFile.delete();
        // 2. 再删除目录. delete 要求必须是空目录才能删除.
        File baseDir = new File(getQueueDir(queueName));
        boolean ok3 = baseDir.delete();
        if (!ok1 || !ok2 || !ok3) {
            throw new IOException("删除队列目录失败! baseDir=" + baseDir.getAbsolutePath());
        }
    }

    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
        if (!checkFilesExists(queue.getName())) {
            throw new MqException("[MessageFileManager] 队列匹配的文件不存在! queueName=" + queue.getName());
        }
        // 1. 先把 message 转成二进制
        byte[] messageBinary = BinaryTool.toBytes(message);
        // 此处的锁对象以队列为维度. 不同队列之间不涉及锁冲突.
        synchronized (queue) {
            // 2. 先获取到文件总长度
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
            message.setOffsetBeg(queueDataFile.length() + 4);
            message.setOffsetEnd(queueDataFile.length() + 4 + messageBinary.length);
            // 3. 写入消息数据文件
            try (OutputStream outputStream = new FileOutputStream(queueDataFile, true)) {
                DataOutputStream dataOutputStream = new DataOutputStream(outputStream);
                // 先写入消息长度
                dataOutputStream.writeInt(messageBinary.length);
                // 再写入消息本体
                dataOutputStream.write(messageBinary);
            }
            // 4. 写入消息统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount += 1;
            stat.validCount += 1;
            writeStat(queue.getName(), stat);
        }
    }

    // 从统计文件中读取结果
    private Stat readStat(String queueName) {
        Stat stat = new Stat();
        try (InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))) {
            Scanner scanner = new Scanner(inputStream);
            stat.totalCount = scanner.nextInt();
            stat.validCount = scanner.nextInt();
            return stat;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    // 向统计文件中写入结果
    private void writeStat(String queueName, Stat stat) {
        try (OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))) {
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.write(stat.totalCount + "\t" + stat.validCount);
            printWriter.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private boolean checkFilesExists(String queueName) {
        File queueData = new File(getQueueDataPath(queueName));
        if (!queueData.exists()) {
            return false;
        }
        File queueStat = new File(getQueueStatPath(queueName));
        if (!queueStat.exists()) {
            return false;
        }
        return true;
    }

    // 把文件上的对应消息给删除掉. (标记成无效)
    public void deleteMessage(MSGQueue queue, Message message) throws IOException, ClassNotFoundException {
        synchronized (queue) {
            try (RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()), "rw")) {
                // 1. 先从文件中读取出 Message 的数据
                byte[] bufferSrc = new byte[(int) (message.getOffsetEnd() - message.getOffsetBeg())];
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.read(bufferSrc);
                // 2. 转成 Message 对象
                Message diskMessage = (Message) BinaryTool.fromBytes(bufferSrc);
                // 3. 设置成无效.
                diskMessage.setIsValid((byte)0x0);
                // 4. 重新写入文件
                byte[] bufferDest = BinaryTool.toBytes(diskMessage);
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.write(bufferDest);
            }
        }
    }

    // 从消息数据文件中读取出所有消息
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws MqException, IOException, ClassNotFoundException {
        // 记录当前读到的数据在文件的 offset
        long currentOffset = 0;
        LinkedList<Message> messages = new LinkedList<>();
        try (InputStream inputStream = new FileInputStream(getQueueDataPath(queueName))) {
            DataInputStream dataInputStream = new DataInputStream(inputStream);
            while (true) {
                // 读到文件末尾, 会触发 EOFException
                int messageSize = dataInputStream.readInt();
                byte[] buffer = new byte[messageSize];
                int actualSize = dataInputStream.read(buffer);
                if (messageSize != actualSize) {
                    throw new MqException("[MessageFileManager] 文件格式错误! queueName=" + queueName);
                }
                Message message = (Message) BinaryTool.fromBytes(buffer);
                if (message.getIsValid() != 0x1) {
                    // 被删除的无效数据, 直接跳过. 不要忘记更新 currentOffset
                    currentOffset += 4 + messageSize;
                    continue;
                }
                // 计算该 message 的 offset
                message.setOffsetBeg(currentOffset + 4);
                message.setOffsetEnd(currentOffset + 4 + messageSize);
                // 每个消息, 开头 4 个字节保存的是消息的长度. 接下来 [offsetBeg, offsetEnd) 是消息体
                currentOffset += 4 + messageSize;
                messages.add(message);
            }
        } catch (EOFException e) {
            // 数据读取完毕, 循环正常退出!
            System.out.println("[MessageFileManager] 恢复 Message 数据完成!");
        }
        return messages;
    }

    // 检查是否要针对文件进行 GC 操作
    public boolean checkGC(String queueName) {
        Stat stat = readStat(queueName);
        if (stat.totalCount >= 2000 && (double)stat.validCount / (double) stat.totalCount <= 0.5) {
            return true;
        }
        return false;
    }

    private String getQueueDataNewPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data_new.txt";
    }

    // 真正执行 GC 操作
    // 使用复制算法.
    // 先创建一个新的文件, 名字为 "queue_data_new.txt"
    // 然后加载出旧的文件的所有有效消息内容
    // 把这些内容写入到新的文件中.
    // 删除旧文件, 对新文件重命名.
    public void gc(MSGQueue queue) throws MqException, IOException, ClassNotFoundException {
        synchronized (queue) {
            long gcBeg = System.currentTimeMillis();
            // 1. 创建一个新的文件, 名字为 "queue_data_new.txt"
            File queueDataNew = new File(getQueueDataNewPath(queue.getName()));
            if (queueDataNew.exists()) {
                throw new MqException("[MessageFileManager] gc 时发现队列新数据文件已经存在! queueName=" + queue.getName());
            }
            boolean ok = queueDataNew.createNewFile();
            if (!ok) {
                throw new IOException("创建文件失败! queueDataNew=" + queueDataNew.getAbsolutePath());
            }
            // 2. 遍历旧文件, 读取出每个对象 (只保留有效消息)
            List<Message> messageList = loadAllMessageFromQueue(queue.getName());
            // 3. 把有效消息写入到新的文件中.
            try (OutputStream outputStream = new FileOutputStream(queueDataNew)) {
                DataOutputStream dataOutputStream = new DataOutputStream(outputStream);
                for (Message message : messageList) {
                    byte[] buffer = BinaryTool.toBytes(message);
                    dataOutputStream.writeInt(buffer.length);
                    dataOutputStream.write(buffer);
                }
            }
            // 4. 删除 queue_data.txt, 把 queue_data_new.txt 重命名为 queue_data
            File queueDataOld = new File(getQueueDataPath(queue.getName()));
            ok = queueDataOld.delete();
            if (!ok) {
                throw new IOException("删除文件失败! queueDataOld=" + queueDataOld.getAbsolutePath());
            }
            ok = queueDataNew.renameTo(queueDataOld);
            if (!ok) {
                throw new IOException("文件重命名失败! queueDataOld=" + queueDataOld.getAbsolutePath() +
                        ", queueDataNew=" + queueDataNew.getAbsolutePath());
            }
            long gcEnd = System.currentTimeMillis();
            System.out.println("[MessageFileManager] gc 执行完毕! queueName=" + queue.getName() + ", time=" + (gcEnd - gcBeg) + "ms");
        }
    }
}
